#include "rpc_proxy.h"
#include "rpc_logger.h"
#include "rpc_root.h"

namespace rpc {

std::uint32_t ProxyCreatorImpl::proxyID_ = 1;

ProxyCreatorImpl::~ProxyCreatorImpl() { factoryMap_.clear(); }

ProxyPtr ProxyCreatorImpl::getService(ServiceUUID serviceUUID,
                                      TransportPtr &trans, bool attached,
                                      Rpc *rpc) {
  if (!trans) {
    return nullptr;
  }
  auto it = factoryMap_.find(serviceUUID);
  if (it == factoryMap_.end()) {
    return nullptr;
  }
  ProxyPtr proxy = it->second->create(rpc);
  if (!proxy) {
    return nullptr;
  }
  auto proxyImpl = std::dynamic_pointer_cast<ProxyImpl>(proxy);
  if (!proxyImpl) {
    return nullptr;
  }
  proxyImpl->setID(proxyID_++);
  proxyImpl->setTransport(trans);
  if (attached) {
    proxyImpl->attach();
  }
  getProxyManager(rpc)->add(proxy);
  return proxy;
}

ProxyPtr ProxyCreatorImpl::getService(const std::string &className,
                                      TransportPtr &trans, bool attached,
                                      Rpc *rpc) {
  auto it = nameIdMap_.find(className);
  if (it == nameIdMap_.end()) {
    return nullptr;
  }
  return getService(it->second, trans, attached, rpc);
}

void ProxyCreatorImpl::addFactory(ServiceUUID serviceUUID,
                                  ProxyFactory *factory) {
  factoryMap_.emplace(serviceUUID, factory);
  nameIdMap_.emplace(factory->name(), serviceUUID);
}

ProxyID ProxyCreatorImpl::createProxyID() { return proxyID_++; }

void ProxyCreatorImpl::cleanup() {
  factoryMap_.clear();
  nameIdMap_.clear();
}

ProxyManagerImpl::ProxyManagerImpl(RpcImpl *rpc) { rpc_ = rpc; }

ProxyManagerImpl::~ProxyManagerImpl() {}

void ProxyManagerImpl::add(ProxyPtr &proxy) {
  proxyMap_[std::dynamic_pointer_cast<ProxyImpl>(proxy)->getID()] = proxy;
}

void ProxyManagerImpl::destroy(ProxyID id) {
  auto it = proxyMap_.find(id);
  if (it == proxyMap_.end()) {
    return;
  }
  proxyMap_.erase(it);
  dynamic_cast<ProxyCallManagerImpl *>(rpc_->getProxyCallManager())
      ->removeAll(id);
}

ProxyPtr ProxyManagerImpl::get(ProxyID id) {
  auto it = proxyMap_.find(id);
  if (it == proxyMap_.end()) {
    return nullptr;
  }
  if (it->second.expired()) {
    proxyMap_.erase(it);
    return nullptr;
  }
  return it->second.lock();
}

void ProxyManagerImpl::cleanup() {
  auto temp = proxyMap_;
  for (auto it = temp.begin(); it != temp.end(); it++) {
    if (it->second.expired()) {
      continue;
    }
    auto weak_pry = it->second;
    destroy(it->first);
    std::dynamic_pointer_cast<ProxyImpl>(weak_pry.lock())->setRpc(nullptr);
  }
  temp.clear();
  proxyMap_.clear();
}

ProxyCall::~ProxyCall() {
  stopExecTimer();
  if (data_) {
    rpc_->free(data_);
  }
  data_ = nullptr;
}

const std::string &ProxyCall::getCallName() {
#ifdef DEBUG
  return callName_;
#else
  static std::string unknown("unknown");
  return unknown;
#endif
}

CallID ProxyCall::getCallID() { return callID_; }

ErrorID ProxyCall::getErrorID() { return errorID_; }

bool ProxyCall::isCoroutinized() { return inCoro_; }

void ProxyCall::setCoro() { inCoro_ = true; }

void ProxyCall::setErrorID(ErrorID errorID) { errorID_ = errorID; }

void ProxyCall::setCallID(CallID callID) { callID_ = callID; }

void ProxyCall::setTimerHandle(TimerHandle handle) { timerHandle_ = handle; }

void ProxyCall::setTimeout(int timeout) { timeout_ = timeout; }

int ProxyCall::getTimeout() { return timeout_; }

void ProxyCall::setRetry(int retry) { retry_ = retry; }

int ProxyCall::getRetry() { return retry_; }

bool ProxyCall::needRetry() { return (retry_ > 0); }

TimerHandle ProxyCall::getTimerHandle() { return timerHandle_; }

void ProxyCall::setNoExcept(bool noExcept) { noExcept_ = noExcept; }

bool ProxyCall::getNoExcept() { return noExcept_; }

Rpc *ProxyCall::getRpc() { return rpc_; }

void ProxyCall::setTraceID(const std::string &traceID) { traceID_ = traceID; }

const std::string &ProxyCall::getTraceID() { return traceID_; }

void ProxyCall::setSpanID(std::uint64_t spanID) { spanID_ = spanID; }

std::uint64_t ProxyCall::getSpanID() { return spanID_; }

void ProxyCall::setParentSpanID(std::uint64_t parentSpanID) {
  parentSpanID_ = parentSpanID;
}

std::uint64_t ProxyCall::getParentSpanID() { return parentSpanID_; }

rpc::TransportPtr ProxyCall::getTransport() { return transport_; }

std::time_t ProxyCall::getForkTimestamp() { return forkTimestamp_; }

void ProxyCall::setForkTimestamp(std::time_t forkTimestamp) {
  forkTimestamp_ = forkTimestamp;
}

void ProxyCall::resend() {
  if (transport_ && !transport_->isClose()) {
    transport_->send(data_, dataSize_);
  }
}

void ProxyCall::setData(const char *data, int size) {
  if (data_) {
    rpc_->free(data_);
  }
  data_ = (char *)rpc_->malloc(size);
  memcpy(data_, data, size);
  dataSize_ = size;
}

void ProxyCall::stopExecTimer() {
  if (!timerHandle_) {
    return;
  }
  if (timerHandle_ || getRpc()->getProxyCallManager()) {
    dynamic_cast<ProxyCallManagerImpl *>(getRpc()->getProxyCallManager())
        ->getTimer()
        ->cancelTimer(timerHandle_);
  }
  timerHandle_ = nullptr;
}

void ProxyCall::startExecTimer() {
  if (timerHandle_) {
    dynamic_cast<ProxyCallManagerImpl *>(getRpc()->getProxyCallManager())
        ->getTimer()
        ->cancelTimer(timerHandle_);
    timerHandle_ = nullptr;
  }
  timerHandle_ =
      dynamic_cast<ProxyCallManagerImpl *>(getRpc()->getProxyCallManager())
          ->getTimer()
          ->addTimerOnce(
              [&](std::uint64_t callID) {
                auto callPtr = getRpc()->getProxyCallManager()->get(
                    static_cast<CallID>(callID));
                if (!callPtr) {
                  return;
                }
                if (callPtr->needRetry()) {
                  getRpc()->getProxyCallManager()->retry(callPtr);
                } else {
                  getRpc()->getProxyCallManager()->timeout(callPtr);
                }
              },
              getTimeout(), getCallID());
}

ProxyImpl::~ProxyImpl() {
  if (rpc_) {
    rpc_->getProxyManager()->destroy(getID());
  }
}

ServiceID ProxyImpl::getTarget() { return targetID_; }

void ProxyImpl::setTarget(ServiceID serviceID) { targetID_ = serviceID; }

TransportPtr ProxyImpl::getTransport() { return transport_; }

void ProxyImpl::setTransport(TransportPtr &transport) {
  transport_ = transport;
}

void ProxyImpl::attach() { attached_ = true; }

void ProxyImpl::detach() { attached_ = false; }

bool ProxyImpl::isAttached() { return attached_; }

void ProxyImpl::setID(ProxyID id) { id_ = id; }

void ProxyImpl::setGlobalIndex(GlobalIndex globalIndex) {
  globalIndex_ = globalIndex;
}

bool ProxyImpl::isConnected() {
  if (!transport_) {
    return false;
  }
  return !transport_->isClose();
}

void ProxyImpl::setRpc(Rpc *rpc) { rpc_ = dynamic_cast<RpcImpl *>(rpc); }

ProxyID ProxyImpl::getID() { return id_; }

GlobalIndex ProxyImpl::getGlobalIndex() { return globalIndex_; }

Rpc *ProxyImpl::getRpc() { return rpc_; }

void ProxyCall::setCallName(const std::string &name) {
#ifdef DEBUG
  callName_ = name;
#else
  (void)name;
#endif
}

void ProxyCallManagerImpl::destroy(CallID callID) {
  auto it = callMap_.find(callID);
  if (it == callMap_.end()) {
    Logbegin << "ProxyCallManagerImpl::destroy, Proxy call not found[" << callID
             << "]" << Logend;
    return;
  }
  it->second->stopExecTimer();
  callMap_.erase(it);
}

void ProxyCallManagerImpl::update(std::time_t ms) {
  while (timer_.runOnce(ms)) {
  }
}

ProxyCallPtr ProxyCallManagerImpl::get(CallID callID) {
  auto it = callMap_.find(callID);
  if (it == callMap_.end()) {
    return nullptr;
  }
  return it->second;
}

ProxyCallManagerImpl::ProxyCallManagerImpl(RpcImpl *rpc) { rpc_ = rpc; }

ProxyCallManagerImpl::~ProxyCallManagerImpl() {}

void ProxyCallManagerImpl::add(ProxyCallPtr call) {
  call->setCallID(callID_++);
  callMap_.emplace(call->getCallID(),
                   std::dynamic_pointer_cast<ProxyCall>(call));
  call->startExecTimer();
}

void ProxyCallManagerImpl::retry(ProxyCallPtr callPtr) {
  Logbegin << "Retry method name [" << callPtr->getCallName() << "]" << Logend;
  callPtr->startExecTimer();
  callPtr->setRetry(callPtr->getRetry() - 1);
  callPtr->setErrorID(static_cast<ErrorID>(RpcError::TIMEOUT));
  callPtr->resend();
}

void ProxyCallManagerImpl::cleanup() { timer_.clear(); }

void ProxyCallManagerImpl::removeAll(ProxyID proxyID) {
  for (auto it = callMap_.begin(); it != callMap_.end();) {
    if (it->second->getProxyID() == proxyID) {
      it = callMap_.erase(it);
    } else {
      it++;
    }
  }
}

void ProxyCallManagerImpl::timeout(ProxyCallPtr &callPtr) {
  Logbegin << "Proxy call timeout [" << callPtr->getCallName() << "]"
           << ", CallID [" << callPtr->getCallID() << "], Timeout["
           << callPtr->getTimeout() << "]" << Logend;
  callPtr->setErrorID(static_cast<ErrorID>(RpcError::TIMEOUT));
  callPtr->doRet(0);
}

TimerBase *ProxyCallManagerImpl::getTimer() { return &timer_; }

ProxyManager *getProxyManager(Rpc *rpc) { return rpc->getProxyManager(); }

ProxyCreator &getGlobalProxyCreator() {
  static ProxyCreatorImpl globalProxyCreator;
  return globalProxyCreator;
}

void cleanup_all_proxy_stuff() {
  dynamic_cast<ProxyCreatorImpl &>(getGlobalProxyCreator()).cleanup();
}

} // namespace rpc
